package delivery

import (
	"github.com/jmoiron/sqlx"
	"learning/gooop/saga/mqs/database"
	"learning/gooop/saga/mqs/eventbus"
	"learning/gooop/saga/mqs/logger"
	"learning/gooop/saga/mqs/models"
	"learning/gooop/saga/mqs/models/events"
	"sync"
	"time"
)

type tDeliveryService struct {
	rwmutex *sync.RWMutex
	workers map[string]*tDeliveryWorker
}

func newDeliveryService() *tDeliveryService {
	it := new(tDeliveryService)
	it.init()
	return it
}

func (me *tDeliveryService) init() {
	me.rwmutex = new(sync.RWMutex)
	me.workers = make(map[string]*tDeliveryWorker)
}

func (me *tDeliveryService) handleBootEvent(_ string, _ interface{}) {
	go me.beginCreatingWorkers()
	go me.beginCleanExpiredWorkers()
}


func (me *tDeliveryService) beginCreatingWorkers() {
	logger.Logf("tDeliveryService.beginCreatingWorkers")
	for {
		e := database.DB(func(db *sqlx.DB) error {
			now := time.Now().UnixNano()
			rows, err := db.Queryx("select ClientID, NotifyURL, ExpireTime from subscriber where ExpireTime>?", now)
			if err != nil {
				return err
			}

			defer rows.Close()
			for rows.Next() {
				it := new(tWorkerInfo)
				err = rows.StructScan(it)
				if err != nil {
					return err
				}

				me.createWorker(it)
			}

			return nil
		})

		if e != nil {
			logger.Logf("tDeliveryService.beginCreatingWorkers, error = %s", e.Error())
		}
		time.Sleep(time.Duration(5) * time.Second)
	}
}


func (me *tDeliveryService) beginCleanExpiredWorkers() {
	logger.Logf("tDeliveryService.beginCleanExpiredWorkers")
	for range time.Tick(time.Duration(30) * time.Second) {
		me.clean()
	}
}

func (me *tDeliveryService) clean() {
	logger.Logf("tDeliveryService.clean")
	me.rwmutex.RLock()
	var keys []string
	for k,v := range me.workers {
		if v.isExpired() {
			keys = append(keys, k)
		}
	}
	me.rwmutex.RUnlock()

	if len(keys) == 0 {
		return
	}

	me.rwmutex.Lock()
	defer me.rwmutex.Unlock()
	for _,k := range keys {
		delete(me.workers, k)
	}
}

func (me *tDeliveryService) handleSubscriberRegistered(_ string, args interface{}) {
	it, ok := args.(*models.SubscribeMsg)
	if !ok {
		return
	}

	me.createWorker(&tWorkerInfo{
		it.ClientID, it.NotifyUrl, it.ExpireTime,
	})
}

func maxInt64(a, b int64) int64 {
	if a >= b {
		return a
	}
	return b
}

// createWorker: to create new delivery worker by client id
func (me *tDeliveryService) createWorker(info *tWorkerInfo) {
	me.rwmutex.RLock()
	w,ok := me.workers[info.ClientID]
	me.rwmutex.RUnlock()
	if ok {
		w.info.ExpireTime = maxInt64(w.info.ExpireTime, info.ExpireTime)
		return
	}

	me.rwmutex.Lock()
	defer me.rwmutex.Unlock()
	w,ok = me.workers[info.ClientID]
	if ok {
		w.info.ExpireTime = maxInt64(w.info.ExpireTime, info.ExpireTime)
		return
	}
	me.workers[info.ClientID] = newDeliveryWorker(info)
}

// init: hook event handlers
func init() {
	eventbus.GlobalEventBus.Sub(events.SystemBootEvent,
		"gDeliveryService.handleBootEvent", gDeliveryService.handleBootEvent)
	eventbus.GlobalEventBus.Sub(events.SubscriberRegisteredEvent,
		"gDeliveryService.handleSubscriberRegistered", gDeliveryService.handleSubscriberRegistered)
}

// gDeliveryService: global singleton for managing delivery workers
var gDeliveryService = newDeliveryService()